package com.kafka.xiang.controller;

import com.kafka.xiang.agentcontrol.ControlAgent;
import com.kafka.xiang.common.TestAgentServerInfo;
import com.kafka.xiang.config.Configuration;
import com.kafka.xiang.kakfamodual.KafkaConsumerOperation;
import com.kafka.xiang.kakfamodual.KafkaListener;
import org.apache.kafka.common.protocol.types.Field;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.beans.factory.annotation.Autowired;

/**
 *
 *
 * @author xiang
 */
@Controller
@SuppressWarnings("all")
@RequestMapping("/api/kafka/")
public class ProducerController {
    private final static Logger LOG = LoggerFactory.getLogger(ProducerController.class);
    private boolean consumerCreationSuccessful = false;
    private ExecutorService executor = Executors.newSingleThreadExecutor();//线程管理
    @Autowired
    private Configuration configuration;
    private int tryCount=1;//连接次数
    KafkaConsumerOperation operation = new KafkaConsumerOperation();

    @GetMapping("AgentInit")
    @ResponseBody
    public String AentInit(@RequestParam(name = "kafkaServer",defaultValue = "127.0.0.1:9092", required = true) String kafkaServer,
                        @RequestParam(name = "listeningTopic", defaultValue = "topic-PA",required = true) String listeningTopic,
                        @RequestParam(name = "consumerTopic", defaultValue = "topic-AP,topic-AC",required = true) String consumerTopic,
                        @RequestParam(name = "groupId", defaultValue = "G1",required = true) String groupId) {
        kafkaServer = configuration.getKafkaServer();//公共的kafkaserver，方便各个模块相互发送监听消息。
        System.out.println(kafkaServer+listeningTopic+consumerTopic+groupId);
        LOG.info(kafkaServer);
        TestAgentServerInfo info = new TestAgentServerInfo(kafkaServer, listeningTopic, consumerTopic);
        System.out.println(consumerTopic.split(","));
        String[] c_topic =consumerTopic.split(",");
        for (String topic : c_topic) {
            System.out.println(topic);
        }
        ControlAgent agent=new ControlAgent(c_topic,kafkaServer);
        operation.setMessageHandler(agent::handler);//// 将agent对象的方法handler赋给operation对象的messageHandler接口
        //python生成KafkaListener实例时只用把operation::createConsumer方法放到参数中就可以了,在 Java 中，可以使用方法引用来传递方法作为参数,方法引用的语法是 对象::方法
        //KafkaListener 构造函数中期望的 Runnable 接口一致（不接受参数），则可以使用方法引用直接传递
        //如果签名不一致(接受参数)，你需要使用 lambda 表达式来适配
        System.out.println(info.getListeningTopic()+info.getKafkaServer()+groupId);
        Runnable listeningFun = () -> operation.createConsumer(info.getListeningTopic(), info.getKafkaServer(), groupId, true);
        Runnable terminateFun = () -> operation.terminateConsumer();
        KafkaListener listener = new KafkaListener(info.getKafkaServer(), info.getListeningTopic(), groupId,listeningFun,terminateFun);
        //先订阅一次，看是否成功，不成功再进行重新订阅。
        listener.startListening();
        try {
            // 休眠waitingTime秒,等待订阅
            Thread.sleep( 3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
            // 处理中断异常，如果需要的话
        }
        consumerCreationSuccessful=operation.getconsumerCreationSuccessful();
        System.out.println("先订阅一次，看是否成功，不成功再进行重新订阅。");
        System.out.println(consumerCreationSuccessful);
        while (!consumerCreationSuccessful && tryCount <= configuration.getConnectCount()) {
            System.out.println(String.format("正在尝试第%d次重新连接",tryCount));
            System.out.println(consumerCreationSuccessful);
            listener.startListening();//如果在while中，会启动多个adminClient。一直持续监听一段时间，才报错。
            consumerCreationSuccessful=operation.getconsumerCreationSuccessful();
            System.out.println(consumerCreationSuccessful);

            try {
                // 休眠waitingTime秒
                Thread.sleep(configuration.getWaitingTime() * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
                // 处理中断异常，如果需要的话
            }
            //listener.stopListening();
            tryCount++;
        }
        if(consumerCreationSuccessful) return "message:success";
        else return "message:fail";
    }
    @GetMapping("checkout")
    @ResponseBody
    public int[] checkout(){
        int intListeningActive=0;
        int[] messages = new int[4];
        if(consumerCreationSuccessful) intListeningActive=1;
        messages[0]=this.tryCount;
        messages[1]= configuration.getConnectCount();
        messages[2]=configuration.getWaitingTime();
        messages[3]=intListeningActive;
        return messages;
    }

    @GetMapping("stoplistening")
    @ResponseBody
    public boolean stoplistening(){
        System.out.println("停止监听");
        boolean stoplistening =false;
        stoplistening=operation.setActive(stoplistening);
        consumerCreationSuccessful=false;
        return stoplistening;
    }


    @GetMapping("test")
    @ResponseBody
    public String test(){
        System.out.println("hello world!");
        return "ok";
    }
}
